package cn.edu.flink.scala.tutorial.checkpoint

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala._

object CheckpointTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    // 配置重启策略
//     env.setRestartStrategy(new RestartStrategies.FixedDelayRestartStrategyConfiguration(3, Time.seconds(5)))

    // 开启checkpoint
    env.enableCheckpointing(1000L)
    val checkpointConfig = env.getCheckpointConfig

    checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // checkpoint 最小时间间隔
    checkpointConfig.setMinPauseBetweenCheckpoints(1000L)
    // checkpoint 同时checkpoint数量
    checkpointConfig.setMaxConcurrentCheckpoints(1)
    // 外部持久化数据清理方式
    checkpointConfig.setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 不对齐检查点的保存方式
    checkpointConfig.enableUnalignedCheckpoints()
    // 配置持久化位置 FileSystemCheckpointStorage JobManagerCheckpointStorage
    checkpointConfig.setCheckpointStorage("file:///D:/tmp")

    val clickStream = env.addSource(new ClickSource).assignAscendingTimestamps(_.timestamp)

    clickStream
      .map((_,1L))
      .keyBy(_._1)
      .sum(1)
      .print()

    env.execute("CheckpointTest")

  }
}
